package com.zhang.hadoop.flink.test9;

import com.zhang.hadoop.flink.base.ClickSource;
import com.zhang.hadoop.flink.base.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author: zhang yufei
 * @createTime:2022/9/24 11:52
 * @description:
 */
public class SimpleTableExample {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //1.读取数据，得到DataStream
        SingleOutputStreamOperator<Event> eventStream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override
                            public long extractTimestamp(Event event, long l) {
                                return event.timestamp;
                            }
                        }));

        //2.创建执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //3.将DataStream转换成Table
        Table eventTable = tableEnv.fromDataStream(eventStream);

        //4.直接写SQL进行转换
        Table resultTable = tableEnv.sqlQuery("select user,url from " + eventTable);

        //5.基于Table直接转换
        Table resultTable2 = eventTable.select($("user"), $("url"))
                .where($("user").isEqual("欧阳慧"));

        tableEnv.toDataStream(resultTable).print("result1");
        tableEnv.toDataStream(resultTable2).print("result2");

        //7.聚合转换
        tableEnv.createTemporaryView("clickTable",eventTable );
        Table aggResult = tableEnv.sqlQuery("select user,count(url) as cnt from clickTable group by user");

        tableEnv.toChangelogStream(aggResult).print("agg");

        env.execute();
    }
}
